package com.boarsoft.message.rocketmq;

import java.util.List;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.boarsoft.message.core.MessageService;
import com.boarsoft.message.rocketmq.converter.RocketMsgConverter;

public class RocketMQConsumerEx extends RocketMQConsumer {
	private static final Logger log = LoggerFactory.getLogger(RocketMQConsumerEx.class);

	/** */
	protected MessageService handler;
	/** */
	protected RocketMsgConverter converter;

	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		if (handler != null && converter != null) {
			for (MessageExt me : msgs) {
				try {
					log.info("Get message from rocketmq {}", me);
					com.boarsoft.message.bean.Message m = converter.convert(me);
					handler.put(m);
				} catch (Exception e) {
					log.error("Error on handle RocketMQ message", e);
				}
			}
		}
		if (listener != null) {
			listener.consumeMessage(msgs, context);
		}
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}

	public MessageService getHandler() {
		return handler;
	}

	public void setHandler(MessageService handler) {
		this.handler = handler;
	}

	public RocketMsgConverter getConverter() {
		return converter;
	}

	public void setConverter(RocketMsgConverter converter) {
		this.converter = converter;
	}
}
